﻿using System;
using System.Diagnostics;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Threading;
using Sbcx.Collections.Generic;

namespace Sbcx.ComponentModel {

    internal sealed class PersistedConnectableObservable<T> : IPersistedConnectableObservable<T> {

        readonly ConnectedSubject<T> _connectedSubject;
        readonly Subject<ConnectionStatus> _statusSubject = new Subject<ConnectionStatus> ();
        readonly IObservable<ConnectionStatus> _statusObservable;
        readonly Subject<Exception> _exception = new Subject<Exception> ();
        readonly IObservable<Exception> _exceptionObservable;
        readonly Func<IConnectableObservable<T>> _createNewConnectedObservableInstance;
        TimeSpan? _retryInterval;
        readonly IScheduler _scheduler;

        public PersistedConnectableObservable (
            Func<IConnectableObservable<T>> createNewConnectedObservableInstance,
            TimeSpan? retryInterval,
            IScheduler scheduler) {

            _connectedSubject = new ConnectedSubject<T> (OpenConnection, new Subject<T> ());
            _statusObservable = _statusSubject.AsObservable ();
            _exceptionObservable = _exception.AsObservable ();

            _createNewConnectedObservableInstance = createNewConnectedObservableInstance;
            _retryInterval = retryInterval;
            _scheduler = scheduler;
        }

        public IDisposable Connect () {
            return _connectedSubject.Connect ();
        }

        public IDisposable Subscribe (IObserver<T> observer) {
            return _connectedSubject.Subscribe (observer);
        }

        public IObservable<ConnectionStatus> Status {
            get {
                return _statusObservable;
            }
        }

        public IObservable<Exception> Exception {
            get {
                return _exceptionObservable;
            }
        }

        void ScheduleReconnect (TimeSpan? dueTime, Action connect, CancellationToken cancellationToken) {
            IDisposable scheduledReconnect = null;
            scheduledReconnect = dueTime.HasValue ?
                _scheduler.Schedule (dueTime.Value, connect) :
                _scheduler.Schedule (connect);

            cancellationToken.Register (scheduledReconnect.Dispose);
        }

        IDisposable OpenConnection (Action cancel) {
            var cancellationDisposable = new CancellationDisposable ();
            var cancellationToken = cancellationDisposable.Token;
            Action connect = () => { };
            Action disconnect = () => { };
            IConnectableObservable<T> observable;

            // resets the mutable lazy when disposed
            cancellationToken.Register (cancel);

            connect = () => {
                var stopwatch = Stopwatch.StartNew ();
                try {
                    observable = _createNewConnectedObservableInstance ();

                    observable.Subscribe (_connectedSubject.OnNext, exception => {
                        _exception.OnNext (exception);
                        disconnect ();
                        connect ();
                    }, () => {
                        disconnect ();
                        connect ();
                    });

                    _statusSubject.OnNext (ConnectionStatus.Connecting);
                    var connection = observable.Connect ();
                    _statusSubject.OnNext (ConnectionStatus.Connected);

                    IDisposable disconnected = Disposable.Empty;
                    disconnect = () => {
                        disconnected.Dispose ();
                        _statusSubject.OnNext (ConnectionStatus.Disconnecting);
                        try {
                            connection.Dispose ();
                        }
                        catch (Exception exception) {
                            _exception.OnNext (exception);
                        }
                        _statusSubject.OnNext (ConnectionStatus.Disconnected);
                    };
                    disconnected = cancellationToken.Register (disconnect);
                }
                catch (Exception exception) {
                    _exception.OnNext (exception);
                    ScheduleReconnect (CalculateRetryInterval (stopwatch.Elapsed), connect, cancellationToken);
                }
            };

            connect ();

            return cancellationDisposable;
        }

        TimeSpan? CalculateRetryInterval (TimeSpan elapsed) {
            if (!_retryInterval.HasValue)
                return null;
            else {
                elapsed = _retryInterval.Value - elapsed;
                return elapsed < TimeSpan.Zero ? (TimeSpan?)null : elapsed;
            }
        }
    }
}